package com.itbaizhan.calllog;


import org.apache.commons.codec.language.bm.PhoneticEngine;
import org.apache.hadoop.conf.Configuration;
import org.apache.hadoop.hbase.*;
import org.apache.hadoop.hbase.client.*;
import org.apache.hadoop.hbase.filter.*;
import org.apache.hadoop.hbase.util.Bytes;
import org.junit.After;
import org.junit.Before;
import org.junit.Test;

import javax.activation.MailcapCommandMap;
import java.io.IOException;
import java.text.ParseException;
import java.text.SimpleDateFormat;
import java.util.*;

/**
 * @Auther: hwh
 * @Date: 2023-01-23 - 01 - 23 - 10:41
 * @Description: com.itbaizhan.calllog
 * @version: 1.0
 */
public class CallLogDemo {
    //命名空间的定义
    private String namespace = "itbaizhan";
    //表名称
    private String tableName = "phone_log";
    //表名称对应的TableName对象
    private TableName tableNameObj;
    //表DDL对象
    private Admin admin;
    //表数据的DML对象
    private Table table;
    //连接对象
    private Connection connection;
    //定义列族名
    private String family="basic";
    private Random random = new Random();
    private SimpleDateFormat simpleDateFormat = new SimpleDateFormat("yyyy-MM-dd HH:mm:ss");

    @Before
    public void before(){
        Configuration configuration = HBaseConfiguration.create();
        configuration.set("hbase.zookeeper.quorum", "node02,node0,node04");
        try {
            connection = ConnectionFactory.createConnection(configuration);
            admin = connection.getAdmin();
            tableNameObj = TableName.valueOf(namespace+":"+tableName);
            table = connection.getTable(tableNameObj);
        } catch (IOException e) {
            e.printStackTrace();
        }
    }
    @After
    public void after() throws IOException {
        if(table!=null){
            table.close();
        }
        if(admin!=null){
            admin.close();
        }
        if(connection!=null){
            connection.close();
        }
    }
    @Test
    public void createTable() throws IOException {
        //1.定义命名空间描述器
        NamespaceDescriptor build = NamespaceDescriptor.create(namespace).build();
        try {
            //2.获取命名空间，如果存在则不抛出异常
            admin.getNamespaceDescriptor(namespace);
        } catch (IOException e) {
            //3.出现异常说明命名空间不存在
            admin.createNamespace(build);
        }
        //4.创建表描述器的Builder对象
        TableDescriptorBuilder tableDescriptorBuilder =
                TableDescriptorBuilder.newBuilder(tableNameObj);
        //5.创建列族的描述其对象
        ColumnFamilyDescriptor familyDescriptor =
                ColumnFamilyDescriptorBuilder.newBuilder(Bytes.toBytes(family)).build();
        //6.将列族描述器添加到表描述器的Builder对象中
        tableDescriptorBuilder.setColumnFamily(familyDescriptor);
        //7.将表描述器的Builder对象转换为表描述器对象
        TableDescriptor tableDescriptor = tableDescriptorBuilder.build();
        //8.判断表是否存在
        if(admin.tableExists(tableNameObj)){
            //9.如果存在先禁用后删除
            admin.disableTable(tableNameObj);
            admin.deleteTable(tableNameObj);
        }
        //10.创建表
        admin.createTable(tableDescriptor);
    }

    /**
     * 生成10个用户的在某一年内的通话记录，每个用户产生1000条通话记录
     * dnum:对方手机好嘛  type:呼叫类型   0主叫 1表示被叫
     * length: 通话时长 date：时间
     */
    @Test
    public void insert() throws Exception {
        //1.定义一个List<Put>
        List<Put> putList = new ArrayList<>();
        //2.循环10次，模拟10个用户
        for(int i = 0;i<10;i++){
            //3.清空putList，防止上次操作的影响
            putList.clear();
            //4.生成当前用户的手机号码
            String phoneNumber = getPhoneNumber("158");
            System.out.println(phoneNumber);
            //5.模拟每个用户的1000条数据
            for(int j=0;j<1000;j++){
                //cf:length=,cf:dnum=,cf:date=,cf:type= 0表示主叫 1表示被叫
                //6生成每一行通话记录的数据
                String dnum = getPhoneNumber("199");
                int length = random.nextInt(200)+1;
                int type = random.nextInt(2);
                String date = getDate(2050);
                //7.rowkey设计  这样的早点打的电话会出现在前面 parse(date).getTime()时间戳
                String rowkey = phoneNumber+"_"+
                        (Long.MAX_VALUE-simpleDateFormat.parse(date).getTime())+i+j;
                //8.创建put对象
                Put put = new Put(Bytes.toBytes(rowkey));
                put.addColumn(Bytes.toBytes("basic"),Bytes.toBytes("dnum"), Bytes.toBytes(dnum));
                put.addColumn(Bytes.toBytes("basic"),Bytes.toBytes("length"), Bytes.toBytes(length));
                put.addColumn(Bytes.toBytes("basic"), Bytes.toBytes("type"), Bytes.toBytes(type));
                put.addColumn(Bytes.toBytes("basic"), Bytes.toBytes("date"), Bytes.toBytes(date));
                //9.将put对象添加putList
                putList.add(put);
            }
            //10.将当前用户的1000挑通话记录提交
            table.put(putList);
        }
    }

    //2050-01-01 0:0:0 - 2050-12-31 23:59:59
    private String getDate(int year) {
        Calendar calendar = Calendar.getInstance();
        //设置时间
        calendar.set(year,0,1);
        //随机月份
        calendar.add(Calendar.MONTH,random.nextInt(12));
        //随机日
        calendar.add(Calendar.DAY_OF_MONTH,random.nextInt(31));
        //小时数随机
        calendar.add(Calendar.HOUR_OF_DAY,random.nextInt(12));
        //获取时间对象
        Date time = calendar.getTime();
        //将时间转换为格式化的字符串并返回
        return simpleDateFormat.format(time);
    }

    /**生成一个手机号码
     * @param prefix:手机号码的前三位
     * @return 生成的手机号码
     */
    private String getPhoneNumber(String prefix) {
        return prefix+String.format("%08d",random.nextInt(99999999));
    }
    /**
     * 15860547748
     * 15830136852
     * 15817083856
     * 15821530409
     * 15878361725
     * 15899337638
     * 15851463510
     * 15895680959
     * 15834301484
     * 15849660104
     *
     * Process finished with exit code 0
     */
    //查询某用户3月份的通话记录
    @Test
    public void scanData() throws Exception {
        //1.定义某用户的手机号码
        String phoneNumber = "15860547748";
        //2.定义startRow 包含
        String startRow = phoneNumber+"_"+
                (Long.MAX_VALUE-simpleDateFormat.parse("2050-04-01 00:00:00").getTime());
        //3.定义stopRow 包含
        String stopRow = phoneNumber+"_"+
                (Long.MAX_VALUE-simpleDateFormat.parse("2050-03-01 00:00:00").getTime());
        //4.创建Scan对象
        Scan scan = new Scan();
        //5.设置起始和结束行
        scan.withStartRow(Bytes.toBytes(startRow));
        scan.withStopRow(Bytes.toBytes(stopRow));
        //6.执行查询
        ResultScanner resultScanner = table.getScanner(scan);
        //7.解析resultScanner
        for(Result result:resultScanner){
            //8.解析result
            Cell[] cells = result.rawCells();
            String rowInfo = "rowkey:"+Bytes.toString(CellUtil.cloneRow(cells[0]));
            rowInfo +=","+Bytes.toString(CellUtil.cloneQualifier(cells[0]))
                    +":"+Bytes.toString(CellUtil.cloneValue(cells[0]));
            rowInfo +=","+Bytes.toString(CellUtil.cloneQualifier(cells[1]))
                    +":"+Bytes.toString(CellUtil.cloneValue(cells[1]));
            rowInfo +=","+Bytes.toString(CellUtil.cloneQualifier(cells[2]))
                    +":"+Bytes.toInt(CellUtil.cloneValue(cells[2]));
            rowInfo +=","+Bytes.toString(CellUtil.cloneQualifier(cells[3]))
                    +":"+Bytes.toInt(CellUtil.cloneValue(cells[3]));
            //9输出
            System.out.println(rowInfo);
        }
    }
    @Test
    public void deleteCell() throws IOException {
        Delete delete = new Delete(Bytes.toBytes("15860547748_92233695045628298070935"));
        //指定具体的列
        delete.addColumn(Bytes.toBytes("basic"),Bytes.toBytes("length"));
        //执行删除操作
        table.delete(delete);
    }
    /**添加单元格数据演示
     * 添加前查询
     *hbase(main):007:0> get 'itbaizhan:phone_log','15860547748_92233695045628298070935'
     * COLUMN                   CELL
     *  basic:date              timestamp=1638794274438, value=2050-03-30 00:52:53
     *  basic:dnum              timestamp=1638794274438, value=19916554511
     *  basic:type              timestamp=1638794274438, value=\x00\x00\x00\x01
     * 1 row(s)
     * Took 0.0342 seconds
     * 添加后查询
     * hbase(main):008:0> get 'itbaizhan:phone_log','15893647910_92233695046692028070231'
     * COLUMN                   CELL
     *  basic:date              timestamp=1638794274438, value=2050-03-30 00:52:53
     *  basic:dnum              timestamp=1638794274438, value=19916554511
     *  basic:length            timestamp=1638797386409, value=\x00\x00\x00c
     *  basic:type              timestamp=1638794274438, value=\x00\x00\x00\x01
     * 1 row(s)
     * Took 0.0451 seconds
     */
    @Test
    public void insertCell() throws IOException {
        Put put = new Put(Bytes.toBytes("15860547748_92233695045628298070935"));
        put.addColumn(Bytes.toBytes("basic"),Bytes.toBytes("length"),Bytes.toBytes(99));
        table.put(put);
    }

    /**删除行操作
     * 删除前查询
     * hbase(main):008:0> get 'itbaizhan:phone_log','15860547748_92233695045628298070935'
     * COLUMN                   CELL
     *  basic:date              timestamp=1638794274438, value=2050-03-30 00:52:53
     *  basic:dnum              timestamp=1638794274438, value=19916554511
     *  basic:length            timestamp=1638797386409, value=\x00\x00\x00c
     *  basic:type              timestamp=1638794274438, value=\x00\x00\x00\x01
     * 1 row(s)
     * Took 0.0451 seconds
     * 删除后查询
     * hbase(main):009:0> get 'itbaizhan:phone_log','15860547748_92233695045628298070935'
     * COLUMN                   CELL
     * 0 row(s)
     * Took 0.0219 seconds
     * @throws IOException
     */
    @Test
    public void deleteRow() throws IOException {
        Delete delete = new Delete(Bytes.toBytes("15860547748_92233695045628298070935"));
        //执行删除操作
        table.delete(delete);
    }
    public void findByKeyMethodShow(){
        Scan scan = new Scan();
        /**
         * 用于组织两个或多个查询条件之间的关系，参数与如下两个；
         * MUST_PASS_ALL：表示and的关系，交集，同时成立
         * MUST_PASS_ONE：表示or的关系，并集，有一个成立即可
         */
        FilterList filterList = new FilterList(FilterList.Operator.MUST_PASS_ALL);
        //将某一个或多个过滤器添加到filterList中
        /**
         * basic:type=0
         * LESS:小于
         * LESS_OR_EQUAL:小于等于
         * EQUAL 等于
         * *NOT_EQUAL：不等于
         * GREATER_OR_EQUAL大于登录
         * GREATER：大于
         * NO_OP：no operation
         */
         /*SingleColumnValueFilter valueFilter = new SingleColumnValueFilter(
                Bytes.toBytes("basic"),Bytes.toBytes("type"),//指定列族和列描述符
                CompareFilter.CompareOp.EQUAL,//比较条件 hbase2.0.0以前的版本使用
                Bytes.toBytes(0)//value
        );*/
        SingleColumnValueFilter valueFilter =
                new SingleColumnValueFilter(
                        Bytes.toBytes("basic"),
                        Bytes.toBytes("type"),//指定列族和列描述器
                        CompareOperator.EQUAL,//比较条件 hbase2.0.0 开始使用
                        Bytes.toBytes(0) //value
                );
        //将列值过滤器添加到filterList中
        filterList.addFilter(valueFilter);
        //列族过滤器，这种处理方式比较麻烦，一般不用
        ByteArrayComparable bytecom =
                new ByteArrayComparable(Bytes.toBytes("basic")){

                    @Override
                    public byte[] toByteArray() {
                        return new byte[0];
                    }

                    @Override
                    public int compareTo(byte[] bytes, int i, int i1) {
                        return 0;
                    }
                };
        FamilyFilter familyFilter =
                new FamilyFilter(CompareOperator.EQUAL, bytecom);
        //列族过滤直接使用如下方法，简捷高效
        scan.addFamily(Bytes.toBytes("basic"));
        //指定需要的列
        scan.addColumn(Bytes.toBytes("basic"), Bytes.toBytes("dnum"));
        //rowkey的前缀过滤器
        PrefixFilter prefixFilter = new PrefixFilter(Bytes.toBytes("15812345678"));
        //将prefixFilter添加到过滤器的集合中
        filterList.addFilter(prefixFilter);

        //将过滤器的集合添加到scan中
        scan.setFilter(filterList);

    }

    /**
     * 查找某用户（rowkey以15893647910为前缀），并且type= 0所有通话记录
     * 列族：basic
     * 列：dnum,type,date,length
     */
    public void finByFilter(String prefix,int type) throws IOException {
        Scan scan = new Scan();
        //指定查询的列族 select basic:* from itbaizhan:phone_log
        scan.addFamily(Bytes.toBytes("basic"));
        //指定查询的列
        /**
         * select basic:dnum,basic:type,basic:date,basic:length
         * from itbaizhan:phone_log
         */
        scan.addColumn(Bytes.toBytes("basic"), Bytes.toBytes("dnum"));
        scan.addColumn(Bytes.toBytes("basic"), Bytes.toBytes("type"));
        scan.addColumn(Bytes.toBytes("basic"), Bytes.toBytes("date"));
        scan.addColumn(Bytes.toBytes("basic"), Bytes.toBytes("length"));
        //两个查询条件之间的关系 and，必须同时满足才能查询出来
        FilterList filterList = new FilterList(FilterList.Operator.MUST_PASS_ALL);
        //rowkey 前缀过滤器
        PrefixFilter prefixFilter = new PrefixFilter(Bytes.toBytes(prefix));
        filterList.addFilter(prefixFilter);
        //列值过滤器
        SingleColumnValueFilter valueFilter = new SingleColumnValueFilter(
                Bytes.toBytes("basic"),
                Bytes.toBytes("type"),
                CompareOperator.EQUAL,
                Bytes.toBytes(0)
        );
        filterList.addFilter(valueFilter);
        //为scan设置filterList
        scan.setFilter(filterList);
        //执行查询，并返回结果集
        ResultScanner resultScanner = table.getScanner(scan);
        //解析resultScanner
        for(Result result : resultScanner){
            printMsg(result);
        }
    }
    private void printMsg(Result result){
        //1.解析result
        Cell[] cells = result.rawCells();
        String rowInfo = "rowkey:"+Bytes.toString(CellUtil.cloneRow(cells[0]));
        rowInfo +=","+Bytes.toString(CellUtil.cloneQualifier(cells[0]))
                +":"+Bytes.toString(CellUtil.cloneValue(cells[0]));
        rowInfo +=","+Bytes.toString(CellUtil.cloneQualifier(cells[1]))
                +":"+Bytes.toString(CellUtil.cloneValue(cells[1]));
        rowInfo +=","+Bytes.toString(CellUtil.cloneQualifier(cells[2]))
                +":"+Bytes.toInt(CellUtil.cloneValue(cells[2]));
        rowInfo +=","+Bytes.toString(CellUtil.cloneQualifier(cells[3]))
                +":"+Bytes.toInt(CellUtil.cloneValue(cells[3]));
        //2输出
        System.out.println(rowInfo);

    }

    //这是一个main方法，是程序的入口：
    public static void main(String[] args) throws IOException {
        CallLogDemo callLogDemo = new CallLogDemo();
        callLogDemo.before();
        callLogDemo.finByFilter("15821530409",0);
        callLogDemo.after();
    }
}
